use std::{
    io,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll},
};

use bytes::{Buf, Bytes};
use h2::{server::SendResponse, RecvStream};
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, ReadBuf};

use crate::{
    config::http::site_conf::HttpAddressSiteRuntimeConf,
    http::v2::stream::H2Stream,
    listener::Listener,
    openssl::stream::SslStream,
    stream::Stream,
    utils::{time::get_now_msec, unique_id::UniqueID},
};

pub enum HttpStream {
    Stream(Stream),
    SslStream(SslStream<Stream>),
    H2Stream(H2Stream),
}

#[allow(dead_code)]
pub struct HttpAddressSiteRuntimeConfInfo {
    pub conf: Arc<HttpAddressSiteRuntimeConf>,
    pub valid: bool,
}

#[allow(dead_code)]
pub struct HttpConn {
    pub raw_conn: HttpStream,
    // 当前连接上的请求数
    pub requests: usize,
    // 建联时间
    pub establish_ms: u64,

    pre_buf: Option<Bytes>,

    // server conn config.
    pub listener: Option<Arc<Listener>>,
    pub address_site_conf_info: Option<HttpAddressSiteRuntimeConfInfo>,
}

#[allow(dead_code)]
pub type BoxHttpConn = Box<HttpConn>;

#[allow(dead_code)]
impl HttpConn {
    pub fn new_with_stream(
        stream: Stream,
        listener: Option<Arc<Listener>>,
    ) -> Self {
        HttpConn {
            raw_conn: HttpStream::Stream(stream),
            requests: 0,
            establish_ms: get_now_msec(),

            pre_buf: None,
            listener,
            address_site_conf_info: None,
        }
    }

    pub fn new_with_ssl_stream(
        ssl_stream: SslStream<Stream>,
        listener: Option<Arc<Listener>>,
    ) -> Self {
        HttpConn {
            raw_conn: HttpStream::SslStream(ssl_stream),
            requests: 0,
            establish_ms: get_now_msec(),
            pre_buf: None,
            listener,
            address_site_conf_info: None,
        }
    }

    pub fn new_with_h2(
        recv_req: http::Request<RecvStream>,
        send_resp: SendResponse<Bytes>,
        requests: usize,
        listener: Option<Arc<Listener>>,
    ) -> Self {
        HttpConn {
            raw_conn: HttpStream::H2Stream(H2Stream::new(recv_req, send_resp)),
            requests,
            establish_ms: get_now_msec(),
            pre_buf: None,
            listener,
            address_site_conf_info: None,
        }
    }

    pub fn set_pre_buf(&mut self, buf: Bytes) {
        if buf.is_empty() {
            return;
        }

        self.pre_buf = Some(buf);
    }

    pub fn has_pre_data(&self) -> bool {
        if let Some(pre_data) = &self.pre_buf {
            if !pre_data.is_empty() {
                return true;
            }
        }

        false
    }

    pub fn unique_id(&self) -> UniqueID {
        match &self.raw_conn {
            HttpStream::Stream(stream) => stream.unique_id(),
            HttpStream::SslStream(ssl_stream) => {
                ssl_stream.get_ref().unique_id()
            }
            HttpStream::H2Stream(h2_stream) => h2_stream.unique_id(),
        }
    }

    pub fn as_raw_fd(&self) -> Option<std::os::unix::io::RawFd> {
        match &self.raw_conn {
            HttpStream::Stream(stream) => stream.as_raw_fd(),
            HttpStream::SslStream(ssl_stream) => {
                ssl_stream.get_ref().as_raw_fd()
            }
            HttpStream::H2Stream(_) => None,
        }
    }

    pub async fn write_vectored_all(
        &mut self,
        bufs: &[io::IoSlice<'_>],
    ) -> io::Result<()> {
        let mut res = self.write_vectored(bufs).await?;

        for v in bufs {
            if res > v.len() {
                res -= v.len();
                continue;
            }

            let mut slice = &v[..];
            if res > 0 {
                slice = &v[res..];
                res = 0;
            }

            if slice.is_empty() {
                continue;
            }

            self.write_all(slice).await?;
        }

        Ok(())
    }
}

impl AsyncRead for HttpConn {
    fn poll_read(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        if let Some(pre_buf) = self.pre_buf.as_mut() {
            let to_copy = pre_buf.len().min(buf.remaining());
            if to_copy > 0 {
                buf.put_slice(&pre_buf[0..to_copy]);
                pre_buf.advance(to_copy);
            }

            if pre_buf.is_empty() {
                self.pre_buf = None;
            }

            if to_copy > 0 {
                return Poll::Ready(Ok(()));
            }
        }

        match &mut self.raw_conn {
            HttpStream::Stream(stream) => Pin::new(stream).poll_read(cx, buf),
            HttpStream::SslStream(ssl_stream) => {
                Pin::new(ssl_stream).poll_read(cx, buf)
            }
            HttpStream::H2Stream(_) => {
                unreachable!("The h2stream does not read or write data through the I/O interface")
            }
        }
    }
}

impl AsyncWrite for HttpConn {
    fn poll_write(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<Result<usize, io::Error>> {
        match &mut self.raw_conn {
            HttpStream::Stream(stream) => Pin::new(stream).poll_write(cx, buf),
            HttpStream::SslStream(ssl_stream) => {
                Pin::new(ssl_stream).poll_write(cx, buf)
            }
            HttpStream::H2Stream(_) => {
                unreachable!("The h2stream does not read or write data through the I/O interface")
            }
        }
    }

    fn poll_flush(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Result<(), io::Error>> {
        match &mut self.raw_conn {
            HttpStream::Stream(stream) => Pin::new(stream).poll_flush(cx),
            HttpStream::SslStream(ssl_stream) => {
                Pin::new(ssl_stream).poll_flush(cx)
            }
            HttpStream::H2Stream(_) => {
                unreachable!("The h2stream does not read or write data through the I/O interface")
            }
        }
    }

    fn poll_shutdown(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Result<(), io::Error>> {
        match &mut self.raw_conn {
            HttpStream::Stream(stream) => Pin::new(stream).poll_shutdown(cx),
            HttpStream::SslStream(ssl_stream) => {
                Pin::new(ssl_stream).poll_shutdown(cx)
            }
            HttpStream::H2Stream(_) => {
                unreachable!("The h2stream does not read or write data through the I/O interface")
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use bytes::Bytes;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    use crate::{
        runtime::RuntimeManager, stream::create_mock_stream,
        utils::fixed_buffer::FixedBuffer,
    };

    use super::HttpConn;

    #[test]
    fn test_http_conn_pre_buf() {
        RuntimeManager::init_main_runtime(1);
        RuntimeManager::main_runtime_block_on(async {
            let (mock_client, mock_server) = create_mock_stream();

            let mut mock_client_hc =
                HttpConn::new_with_stream(mock_client, None);
            let mut mock_server_hc =
                HttpConn::new_with_stream(mock_server, None);

            let pre_buf = Bytes::from_static("abcdefghijklmn".as_bytes());
            mock_server_hc.set_pre_buf(pre_buf);

            mock_client_hc.write_all(b"11111111111").await.unwrap();
            mock_client_hc.write_all(b"22222222222").await.unwrap();

            let mut buffer = FixedBuffer::new(1024);

            let write_slice = buffer.write_slice();
            let n = mock_server_hc.read(write_slice).await.unwrap();
            buffer.fill(n);

            let write_slice = buffer.write_slice();
            let n = mock_server_hc.read(write_slice).await.unwrap();
            buffer.fill(n);

            assert_eq!(
                buffer.read_slice(),
                b"abcdefghijklmn1111111111122222222222"
            );
        });
    }
}
